package com.burges.net.tableAPIAndSQL.time

import org.apache.flink.api.common.typeinfo.TypeInformation
import org.apache.flink.streaming.api.datastream.DataStreamSource
import org.apache.flink.streaming.api.{datastream, environment}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.{TableEnvironment, TableSchema, Types}
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.sources.{DefinedProctimeAttribute, StreamTableSource}
import org.apache.flink.types.Row

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/14 
  * 描述      时间概念之ProcessTime指定
  */
object ProcessTimeDEMO {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		val tStreamEnv: StreamTableEnvironment = TableEnvironment.getTableEnvironment(environment)
		// 获取数据集
		val stream: DataStream[(String, String)] = environment.fromElements(("a","flink"))

		/**
		  * 第一种方式：将DataStream数据转换成Table中通过.proctime来指定
		  */
		// 将DataStream数据转换成Table
//		val table = tStreamEnv.fromDataStream(stream, 'id, 'value, 'process_time.proctime)
		// 基于process_time时间属性创建滚动窗口
//		val windowTable = table.window(....)

		/**
		  * 第二种方式的注册
		  */
		// 注册输入数据源
		tStreamEnv.registerTableSource("InputProcessTime", new InputProcessTimeSource)
		// 在窗口中使用输入数据源，并基于TableSource中定义的ProcessTime字段创建窗口

	}

	/**
	  * 第二种方式：通过TableSource函数定义
	  */
	// 定义InputEventSource创建外部数据源
	// 并实现DefinedRowtimeAttributes接口以定义EventTime时间属性
	class InputProcessTimeSource extends StreamTableSource[Row] with DefinedProctimeAttribute{
		// 定义获取DataStream数据集的逻辑
		override def getDataStream(execEnv: environment.StreamExecutionEnvironment): datastream.DataStream[Row] = {
			//定义获取DataStream数据集的逻辑
			val inputStream: DataStreamSource[(String, String)] = execEnv.fromElements(("", ""))
			// ....
			// 将数据集转换成DataStream[ROW]格式
			val stream = inputStream.map(...)
			// 不需要指定watermark信息
			stream
		}

		// 定义TableAPI中的时间属性信息
		override def getProctimeAttribute: String = {
			// 该字段将会被添加到Schema尾部
			"process_time"
		}

		// 定义数据集字段名称和类型
		override def getReturnType: TypeInformation[Row] = {
			val names = Array[String]("id", "value")
			val types = Array[TypeInformation[_]](Types.STRING, Types.STRING)
			Types.ROW(names, types)
		}

		override def getTableSchema: TableSchema = ???
	}

}
